package com.niit.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * Date:2024/10/22
 * Author：Ys
 * Description:
 */
public class WordCountDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf,WordCountDriver.class.getName());

        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job,new Path("input/wordcount"));

        Path outputPath = new Path("output/wordcount");

        outputPath.getFileSystem(conf).delete(outputPath,true);

        FileOutputFormat.setOutputPath(job,outputPath);

        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0:1);


    }

}

/*
   Mapper是用来读文件中的每一行数据，读取到每一行数据后按规则进行拆分处理。
   KEYIN:表示Map阶段输出KV中的K的类型，在默认组件（普通文本文件），表示起始位置偏移量（行号），IntWritable,LongWritable
   VALUEIN: 表示Map阶段KV中的V的类型，表示每一行数据，Text
               K:行号
               V:每一行的文本内容
   KEYOUT:表示Map阶段输出KV中的K的类型，K的类型的是和业务相关，在本需求中输出是单词 所以是Text
   VALUEOUT:表示Map阶段输出KV中的V的类型，V的类型是和业务相关，在本需求中，单词出现的次数，所以是IntWritable

 */
class WordCountMapper extends Mapper<LongWritable, Text,Text,IntWritable> {
    //当执行Mapper阶段的时候，在第一次调用map()方法前会调用setup()方法，仅会调用一次，做初始化
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        super.setup(context);
    }
    //map()方法，是核心方法，每读取一行数据，都会调用一次map()方法 多次调用
    /*
    Hadoop Python
    Python HBase
    Java Hadoop
    HTML Hadoop
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //Mapper输出kv的类型  <单词,1>
        Text keyOut = new Text();
        IntWritable valueOut = new IntWritable(1);
        //将读取的一行内容根据分隔符进行拆分  Hadoop Python
        String[] words = value.toString().split(" ");// ["Hadoop","Python"]
        for (String word:words) {
            keyOut.set(word);
            //输出单词，标记为1，因为每个单词至少出现1次
            context.write(keyOut,valueOut);
        }


    }
    //当执行Mapper阶段的时候，最后一次调用map()方法后会调用cleanup()方法，仅会调用一次，做收尾工作
    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        super.cleanup(context);
    }
}
/*
    KEYIN:表示Reduce阶段输入KV中的K的类型，在本需求中，单词,等于Mapper阶段的输出的KEY
    VALUEIN:表示Reduce阶段输入KV中的V的类型，在本需求中，单词出现的次数，等于Mapper阶段的输出VALUE
    KEYOUT:表示Reduce阶段输出KV中的K的类型，在本需求中，单词
    VALUEOUT:表示Reduce阶段输出KV中的V的类型，在本需求中，单词出现的总次数，IntWritable
 */
class WordCountReducer extends Reducer<Text,IntWritable,Text,IntWritable> {

    /**
     * Q：当map的所有输出数据来到reduce之后该如何调用reduce方法进行处理的呢？
     * 此处是Map阶段的输出
     *  <Hadoop,1>  <Python,1>
     *  <Python,1>   <HBase,1>
     *  <Java,1>    <Hadoop,1>
     *  <HTML,1>    <Hadoop,1>
     *Map阶段输出其实并没有直接输出到reduce，中间会经历shuffle(排序的过程)
     *  第一次排序：根据Key进行字典序排序，所以是升序 a-z
     *  第二次排序：根据Key进行分组
     *      <Hadoop,1,1,1>  <HBase,1> <HTML,1>  <Java,1> <Python,1,1>
     * 分组后，同一组的数据会进入reduce方法，reduce方法会接收到一组数据，一组数据中的数据都是相同的key，
     *                所以reduce方法中的key就是相同的
     */
    @Override           //Hadoop                     1,1,1
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {

        //定义次数结果
        IntWritable result = new IntWritable();
        //统计变量
        int count = 0;
        //遍历一组数组，统计次数，取出该组的所有的value
        for (IntWritable value : values) {
            //所有的value进行累加，就是该单词出现的总次数了
            count += value.get();
        }
        result.set(count);

        context.write(key,result);


    }
}


